package com.it.sink;

import com.it.operator.utils.SourceUtils;
import com.it.pojo.Event;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.ArrayList;
import java.util.HashMap;

/**
 * 存储数据到es然后使用kibana进行查询.
 * GET _search
 * {
 *   "query": {
 *     "match": {
 *       "_index": "clicks"
 *     }
 *   }
 *
 * @author code1997
 */
public class EsSinkDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<Event> eventSource = SourceUtils.getEventSource(executionEnvironment);
        ArrayList<HttpHost> hosts = new ArrayList<>();
        hosts.add(new HttpHost("docker_server", 9200));

        eventSource.addSink(new ElasticsearchSink.Builder<>(hosts, new ElasticsearchSinkFunction<Event>() {
            @Override
            public void process(Event event, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                HashMap<String, String> map = new HashMap<>();
                map.put(event.user, event.toString());
                IndexRequest indexRequest = Requests.indexRequest().index("clicks").source(map);
                requestIndexer.add(indexRequest);
            }
        }).build());
        executionEnvironment.execute();
    }
}
